package sql;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.sql.Timestamp;

/**

 

 */
public class D7_KafakSource_rownumber {


    public static void main(String[] args) {


        Configuration flinkConf = new Configuration();
        flinkConf.setString("rest.port","9091");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(flinkConf);
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        env.enableCheckpointing(10 * 1000);
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        checkpointConfig.setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);

        String genSql = "CREATE TABLE ods_tb ( " +
                " stime STRING," +
                " name STRING," +
                " id BIGINT," +
                " rowtime AS to_timestamp(stime)," +
                " WATERMARK FOR rowtime AS rowtime - interval '10' second" +
                ") WITH ( " +
            "  'connector' = 'kafka'," +
            "  'topic' = 'test'," +
            "  'properties.bootstrap.servers' = 'kafka:9092'," +
            "  'properties.group.id' = 'testGroupa'," +
            "  'scan.startup.mode' = 'latest-offset'," +
            "  'format' = 'json'" +
                ")";

        String print = "CREATE TABLE print (" +
            "    stime STRING, " +
            "    name STRING, " +
            "    id BIGINT " +
            ") WITH (" +
            "     'connector' = 'print'" +
            ")";

        String sql = "insert into print " +
                " SELECT " +
            " stime," +
            " name," +
            " id " +
            "FROM " +
            " (" +
            " SELECT *,ROW_NUMBER() OVER(PARTITION BY name ORDER BY rowtime DESC) as rownum FROM ods_tb" +
            " )  where rownum = 1";

        /**
         * 每个name只取rowtime最大的，结果为回撤流
         * +I[2024-10-10 10:00:00, bb, 1]
         * -U[2024-10-10 10:00:00, bb, 1]
         * +U[2024-10-10 10:01:00, bb, 1]
         * -U[2024-10-10 10:01:00, bb, 1]
         * +U[2024-10-10 10:02:00, bb, 1]
         */

        tableEnv.executeSql(genSql);
        tableEnv.executeSql(print);

        System.out.println(sql);
        tableEnv.executeSql(sql);


        System.out.println(new Timestamp(System.currentTimeMillis()));



    }
}
